package net.fltx.p2pserver;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import net.fltx.p2p.common.P2PClientInfo;
import net.fltx.p2p.common.Packet;
import net.fltx.p2p.common.PacketType;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.*;
import java.util.logging.Logger;

/**
 * @author fltx.net
 * @Date 2021/2/21 16:59
 * @Version 1.0
 */
@Slf4j
public class P2PServer {

    /**
     * 客户端
     */
    ConcurrentMap<SocketChannel, P2PClientInfo> p2pClientMap;


    ServerSocketChannel serverSocket;

    Selector selector;

    ExecutorService executorService = Executors.newFixedThreadPool(16);

    public P2PServer() throws IOException {
        p2pClientMap = new ConcurrentHashMap<>();
        serverSocket = ServerSocketChannel.open();
        serverSocket.configureBlocking(false);
        serverSocket.bind(new InetSocketAddress(8000));
        selector = Selector.open();
        serverSocket.register(selector, SelectionKey.OP_ACCEPT);
    }


    public void start() {
        executorService.execute(new Runnable() {
            @Override
            public void run() {
            while (true) {
                try {
                    selector.select();
                    Set<SelectionKey> keys = selector.selectedKeys();
                    Iterator<SelectionKey> iterator = keys.iterator();
                    while (iterator.hasNext()) {
                        SelectionKey selectionKey = iterator.next();
                        if(selectionKey.isAcceptable()){
                            onAccept(selectionKey);
                        }
                        if(selectionKey.isReadable()){
                            onRead(selectionKey);
                        }
                        keys.remove(selectionKey);
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                    //break;
                }catch (Exception e){
                    e.printStackTrace();
                }

            }
        }});

    }


    public void onAccept(SelectionKey selectionKey) {
        try {
            SocketChannel accept = this.serverSocket.accept();
            if(accept==null) return;
            log.info("有新的连接: "+accept.getRemoteAddress());
            accept.configureBlocking(false);
            accept.register(this.selector, SelectionKey.OP_READ);
            this.executorService.execute(() -> {
                log.info("线程池onAccept");
                P2PClientInfo pClientInfo = new P2PClientInfo();
                InetSocketAddress remoteAddress = null;
                try {
                    remoteAddress = (InetSocketAddress) accept.getRemoteAddress();
                    pClientInfo.setAddress(remoteAddress.getAddress().getHostAddress());
                    pClientInfo.setPort(remoteAddress.getPort());
                    p2pClientMap.put(accept, pClientInfo);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });
        } catch (IOException e) {
            e.printStackTrace();
            //throw new RuntimeException(e);
        }catch (Exception e){
            e.printStackTrace();
        }

    }

    /**
     * 处理来自客户端的消息
     *
     * @param selectionKey
     */
    public void onRead(SelectionKey selectionKey) {
        log.info("开始接受消息");
        selectionKey.interestOps(selectionKey.interestOps()&~SelectionKey.OP_READ);
        SocketChannel channel = (SocketChannel) selectionKey.channel();

        executorService.execute(() -> {
            ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            log.info("线程池onRead");
            int len=0;
            try {
                while ((len=channel.read(buffer))>0) {
                    buffer.flip();
                    outputStream.write(buffer.array(),0,len);
                }
                Socket socket = channel.socket();
                log.info("len:{},outputStream.size:{}",len,outputStream.size());
                if (outputStream.size() == 0 && len==-1) {
                    log.info("关闭 - {}",channel.getRemoteAddress());
                    channel.close();
                    selectionKey.cancel();
                    p2pClientMap.remove(channel);
                    return;
                }
                String msg = outputStream.toString("utf-8");
                log.info("{}消息: {}",channel.getRemoteAddress(),msg);
                Packet packet = JSON.parseObject(msg, Packet.class);
                if(packet!=null)
                switch (packet.getType()) {
                    case CLIENT_REGISTER:
                        onClientRegister(channel, packet);

                        break;
                    case CLIENT_MESSAGE:
                }
                selectionKey.interestOps(selectionKey.interestOps()|SelectionKey.OP_READ);
                selector.wakeup();
            } catch (IOException e) {
                selectionKey.cancel();
                e.printStackTrace();
            }catch (Exception e){
                e.printStackTrace();
                String msg="当前端口仅做socket学习使用,请不要无意义的向此端口发数据,谢谢!" +
                        "\nThe current port is only used for socket test, please don't send data to this port meaninglessly, thank you!";
                try {
                    channel.write(ByteBuffer.wrap(msg.getBytes(StandardCharsets.UTF_8)));
                    channel.close();
                    selectionKey.cancel();
                } catch (IOException ioException) {
                    ioException.printStackTrace();
                }
            }
            finally {
                try {
                    outputStream.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

        });
    }

    public void airOnline(Packet<P2PClientInfo> packet) {
        String o = JSON.toJSONString(packet);
        p2pClientMap.forEach((key, val) -> {
            if (packet.getData().getUuid().equals(val.getUuid())) {
                return;
            }
            try {
                ByteBuffer wrap = ByteBuffer.wrap(o.getBytes(StandardCharsets.UTF_8));
                key.write(wrap);

            } catch (Exception e) {
                try {
                    key.close();
                    p2pClientMap.remove(key);
                } catch (IOException ioException) {
                    ioException.printStackTrace();
                }
            }
        });
    }

    public void sendMessage(SocketChannel channel, Packet packet) {
        String o = JSON.toJSONString(packet);
        try {
            channel.write(ByteBuffer.wrap(o.getBytes(StandardCharsets.UTF_8)));
        } catch (IOException exception) {
            exception.printStackTrace();
        }
    }


    public void airMessage(Packet packet) {
        String o = JSON.toJSONString(packet);
        p2pClientMap.forEach((key, val) -> {
            try {
                ByteBuffer wrap = ByteBuffer.wrap(o.getBytes(StandardCharsets.UTF_8));
                key.write(wrap);

            } catch (Exception e) {

            }
        });
    }

    public void onClientRegister(SocketChannel channel, Packet packet) {
        P2PClientInfo info = p2pClientMap.get(channel);
        info.setUsername(packet.getUsername());
        info.setUuid(packet.getUuid());
        log.info("有新客户端注册" + info);
        /* 将p2p列表发给刚刚注册的p2p */
        Packet<Object> listPacket = new Packet<>();
        listPacket.setType(PacketType.SERVER_LIST);
        List<P2PClientInfo> infos = new ArrayList<>();
        p2pClientMap.forEach((key, val) -> {
            if (val.getUuid() == null || "".equals(val.getUuid()) || val.getUuid().equals(info.getUuid())) return;
            infos.add(val);
        });
        listPacket.setData(infos);
        listPacket.sendTo(channel);

        //将当前在线客户端信息发给其他p2p客户端
        Packet<P2PClientInfo> online = new Packet<>();
        online.setData(info);
        airOnline(online);

    }


}
